package org.sg20.parsename;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

public class ParseNameTrie {

    public static abstract class ParseNameTrieMapper extends Mapper<LongWritable, Text, ParseNameOutputKey, Text> {
        protected TrieTree nametree;

        @Override
        protected void setup(Mapper<LongWritable, Text, ParseNameOutputKey, Text>.Context context) throws IOException {

            Path nameList = new Path(context.getCacheFiles()[0]);
            FileSystem fs = FileSystem.get(context.getConfiguration());
            InputStreamReader isr = new InputStreamReader(fs.open(nameList));


            BufferedReader br = new BufferedReader(isr);
            // read until end of file
            String name;
            while ((name = br.readLine()) != null) {
                // add the name to the trie tree
                nametree.insertWord(name);
            }

            br.close();
        }

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, ParseNameOutputKey, Text>.Context context) throws IOException, InterruptedException {
            String words = value.toString();
            String filename = ((FileSplit) context.getInputSplit()).getPath().getName();

            ParseNameOutputKey outkey = new ParseNameOutputKey(filename, key.get());
            Text outvalue = new Text();

            TrieTree.TrieNode prenode = nametree.getRoot(), curnode = null;

            int start = 0, cur = 0;

            while (start < words.length()) {
                while (cur < words.length() && (curnode = prenode.getChild(words.charAt(cur))) != null) {
                    //match by now and continue
                    prenode = curnode;
                    cur++;
                }
                // Here, if abcdefg, cur is 2, curnode is c, prenode is b.
                if (curnode == null) {
                    if (prenode.isEnd()) {
                        // name hit
                        outvalue.set(words.substring(start, cur));
                        context.write(outkey, outvalue);
                        start = cur;
                    } else {
                        // name miss
                        start += 1;
                        cur = start;
                    }
                    // reset prenode
                    prenode = nametree.getRoot();
                } else {
                    //end of line
                    if (prenode.isEnd()) {
                        outvalue.set(words.substring(start, cur));
                        context.write(outkey, outvalue);
                        //end the loop
                    }
                    break;
                }
                //System.out.println(words.length()-start);
            }
        }
    }

    public static class ParseNameTrieHashMapper extends ParseNameTrieMapper {
        public ParseNameTrieHashMapper() {
            nametree = new TrieTree("HashMap");
        }
    }

    public static class ParseNameTrieTreeMapper extends ParseNameTrieMapper {
        public ParseNameTrieTreeMapper() {
            nametree = new TrieTree("TreeMap");
        }
    }


    public static class ParseNameTrieReducer extends Reducer<ParseNameOutputKey, Text, ParseNameOutputKey, Text> {
        // There is nothing to do for reducer
        @Override
        protected void reduce(ParseNameOutputKey key, Iterable<Text> values, Reducer<ParseNameOutputKey, Text, ParseNameOutputKey, Text>.Context context) throws IOException, InterruptedException {
            StringBuilder str = new StringBuilder();
            for (Text value : values) {
                str.append(value.toString());
                str.append("\t");
            }
            context.write(key, new Text(str.toString()));
        }
    }

}